package com.db.stockexchange.network;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.logging.Logger;

public class Server implements Runnable {


	public class Connection {
		private SelectionKey sk;
		private MessageTransceiver transceiver;

		private Connection(SelectionKey sk) {
			this.sk = sk;
			this.transceiver = new MessageTransceiver();
		}

		public boolean alive() {
			return Server.this.activeConnections.containsKey(sk);
		}
		
		public boolean write(byte[] message) throws IOException {
			if (!alive())
				return false;
			
			Server.this.write(message, sk);
			return true;
		}
	}

	public static final int DEFAULT_PORT = 1027;
	private static final int DEFAULT_SELECT_TIMEOUT = 100;

	private AtomicBoolean running = new AtomicBoolean(false);
	private int port;
	private ServerSocketChannel ssc;
	private Selector selector;
	private Map<SelectionKey, Connection> activeConnections = new HashMap<>();
	private List<RequestListener> requestListeners = new LinkedList<>();
	private Logger logger = Logger.getLogger(Server.class.getName());

	public Server(int port) {
		this.port = port;
	}
	
	public Server() {
		this(DEFAULT_PORT);
	}
	
	public void addRequestListener(RequestListener listener) {
		requestListeners.add(listener);
	}

	public boolean isRunning() {
		return running.get();
	}
	
	public boolean stop() {
		return running.compareAndSet(true, false);
	}
	
	@Override
	public void run() {
		
		if (!running.compareAndSet(false, true)) {
			logger.warning("called run(), but server is already running");
			return;
		}
		
		try {
			logger.info("starting server on port " + port);
			selector = Selector.open();
			ssc = ServerSocketChannel.open();
			ssc.configureBlocking(false);
			ssc.socket().bind(new InetSocketAddress(port));
			ssc.register(selector, SelectionKey.OP_ACCEPT);
			logger.info("waiting for connections");
			
			while (running.get()) {
				selector.select(DEFAULT_SELECT_TIMEOUT);
				Set<SelectionKey> selectedKeys = selector.selectedKeys();
				for (SelectionKey sk : selectedKeys) {
					try {
						if (sk.isAcceptable())
							accept();
						if (sk.isReadable())
							read(sk);
					} catch (IOException e) {
						unset(sk);
					}
				}
				selectedKeys.clear();
			}
		} catch (IOException e) {
			logger.severe("server failure: " + e.getMessage());
			throw new RuntimeException("Server error", e);
		} finally {
			logger.warning("shutting down the server");
			try {
				selector.close();
				ssc.close();
			} catch (IOException e) {
				logger.severe("failed to shut down properly");
			}
			
			activeConnections.clear();
			running.set(false);
		}
	}

	private void accept() throws IOException {
		SocketChannel sc = ssc.accept();
		sc.configureBlocking(false);
		SelectionKey sk = sc.register(selector, SelectionKey.OP_READ);
		activeConnections.put(sk, new Connection(sk));
		logger.fine("new connection received " + sk);
		// TODO add 'connection' event
	}

	private void read(SelectionKey sk) throws IOException {
		SocketChannel channel = (SocketChannel) sk.channel();
		Connection connection = activeConnections.get(sk);
		MessageTransceiver transceiver = connection.transceiver;
		transceiver.receive(channel);

		byte[] message;
		while ((message = transceiver.nextMessage()) != null) {
			logger.fine("new message from selection key " + sk);
			for (RequestListener listener : requestListeners) {
				listener.requestReceived(message, connection);
			}
		}
	}

	private void unset(SelectionKey sk) {
		sk.cancel();
		activeConnections.remove(sk);
		logger.fine("disconnected " + sk);
		// TODO add 'disconnect' event
	}

	private void write(byte[] message, SelectionKey sk) throws IOException {
		if (!activeConnections.containsKey(sk)) {
			logger.warning("can't write to a disconnected client " + sk);
			return;
		}
		
		MessageTransceiver transceiver = activeConnections.get(sk).transceiver;
		SocketChannel channel = (SocketChannel) sk.channel();

		try {
			synchronized (channel) {
				transceiver.transmit(message, channel);
			}
		} catch (IOException e) {
			unset(sk);
			throw new IOException(e);
		}
	}

}
